package com.edu.flink.table;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.io.Serializable;
import java.util.Arrays;

/**
 * <p>
 * FlinkTable案例
 * </p>
 *
 * @author jpge
 * @since 2024-04-22
 */
public class TableTest {

    /**
     * 1. 定义数据结构
     * 2. 设置Flink执行环境
     * 3. 定义FlinkTable执行
     * 4. 定义输入源
     * 5. 将数据转换成Table
     * 6. 执行Table查询操作
     * 7. 执行Flink程序
     */
    public static void main(String[] args) throws Exception {
        // 1. 定义执行环境
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(executionEnvironment);

        // 2. 定义输入源
        DataStreamSource<Order> orderA = executionEnvironment.fromCollection(
                Arrays.asList(
                        new Order(1L, "Book1", 3),
                        new Order(2L, "Book2", 4),
                        new Order(3L, "Book3", 2)
                )
        );
        DataStreamSource<Order> orderB = executionEnvironment.fromCollection(
                Arrays.asList(
                        new Order(4L, "Book4", 3),
                        new Order(5L, "Book5", 2),
                        new Order(6L, "Book6", 1)
                )
        );
        // 3. 将数据流量转换成Flink Table
        Table tableA = streamTableEnvironment.fromDataStream(orderA);
        Table tableB = streamTableEnvironment.fromDataStream(orderB);

        //执行查询操作
        Table result = streamTableEnvironment.sqlQuery("SELECT * from " + tableA + " where amount > 2 union all " +
                "select * from " + tableB + " where amount < 2");

        streamTableEnvironment.toAppendStream(result, Order.class).print("=============").setParallelism(1);


        executionEnvironment.execute();
    }


}
